package org.dbstar.scheduler.job;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.dbstar.scheduler.beans.CompoundJob;
import org.dbstar.scheduler.beans.CompoundJobResult;
import org.dbstar.scheduler.beans.Job;
import org.dbstar.scheduler.beans.JobDefine;
import org.dbstar.scheduler.control.JobController;
import org.dbstar.scheduler.control.JobStatusChangeListener;
import org.dbstar.scheduler.control.exception.DaoException;
import org.dbstar.scheduler.identifier.Identifier;
import org.dbstar.scheduler.utils.JobUtils;


/**
 * 复合Job执行器
 * 
 * @author 代波
 * @since 1.0.0
 * @version 1.0.0
 * @date 2012-5-8
 */
public class CompoundJobRunner extends AbstractJobRunner implements JobStatusChangeListener {
	public CompoundJobRunner(JobController jobController, Job job) {
		super(jobController, job);
	}

	@Override
	protected boolean run(Job job) throws Exception {
		return run((CompoundJob) job, (CompoundJobResult) job.getResult());
	}

	private boolean run(CompoundJob job, CompoundJobResult result) throws DaoException {
		// 检查是否所有子任务都运行结束，结束状态包括：kill,cancel,error和finish
		if (result.isFinished()) {
			return true;
		} else {
			// 注册监听器
			List<Identifier> jobs = new ArrayList<Identifier>();
			Set<Identifier> skipJobs = new HashSet<Identifier>();
			for (JobDefine jd : job.getSubJobs()) {
				if (jd.isSkip()) {
					skipJobs.add(jd.getIdentifier());
				} else {
					jobs.add(jd.getIdentifier());
				}
			}
			Identifier identifiers[] = jobs.toArray(new Identifier[jobs.size()]);
			jobController.registJobStatusChangeListener(this, identifiers, null, (JobStatus[]) null);

			try {
				while (!Thread.currentThread().isInterrupted()) {
					// 找到尚未执行的、无依赖、或者依赖均已满足的子任务，修改其状态为submit
					Set<Identifier> ready = result.getReadyJobs();
					if (!ready.isEmpty()) {
						for (Identifier identifier : result.getReadyJobs()) {
							if (skipJobs.contains(identifier)) {
								// skip任务
								result.updateSubJobResult(identifier, JobStatus.finish, true);
							} else {
								jobController.changeJobStatus(identifier, JobStatus.submit);
							}
						}
						continue;
					}

					if (result.isFinished()) {
						break;
					}
					
					synchronized (this) {
						try {
							// 等待子任务状态更新
							this.wait(10000);
						} catch (InterruptedException e) {
						}
					}

					if (result.isFinished()) {
						break;
					}
				}
				if (Thread.currentThread().isInterrupted()) {
					for (Identifier identifier : result.getInitJobs()) {
						// 任务被kill时，设置所有非skip任务的状态为cancel
						if (!skipJobs.contains(identifier)) {
							jobController.changeJobStatus(identifier, JobStatus.cancel);
						}
					}
				}
			} finally {
				jobController.removeJobStatusChangeListener(this);
			}

			return !Thread.currentThread().isInterrupted();
		}
	}

	@Override
	public void onJobStatusChange(Job job, JobStatus oldStatus, JobStatus newStatus) {
		((CompoundJobResult) getJob().getResult()).updateSubJobResult(job);
		if (JobUtils.isFinished(newStatus)) {
			synchronized (this) {
				this.notify();
			}
		}
	}
}
